package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo4Filter {
  def main(args: Array[String]): Unit = {
    //创建flink环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    val filterDS: DataStream[String] = studentDS.filter(new FilterFunction[String] {
      /**
       * 传如一条数据返回一个布尔值，返回true保留数据，返回false过滤数据
       *
       * @param value ：一行数据
       * @return
       */
      override def filter(value: String): Boolean = {
        val gender: String = value.split(",")(3)
        "女".equals(gender)
      }
    })

    filterDS.print()

    env.execute()
  }

}
